package com.jaws.demo.configuration;


import com.alibaba.druid.pool.DruidDataSource;
import com.jaws.demo.common.batch.CsvBeanValidator;
import com.jaws.demo.common.batch.CsvItemProcessor;
import com.jaws.demo.common.batch.CsvJobListener;
import com.jaws.demo.vo.AccountInfo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.validator.Validator;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;

/**
 * spring batch demo configuration
 *
 * @author liangqm
 * @date 2018-04-24 15:28:17
 * @since v1.0.1
 */

@EnableBatchProcessing
@Configuration
//@EnableConfigurationProperties(JawsBatchConfigProperties.class)
@Slf4j
public class JawsBatchConfiguration {



	/**
	 * spring batch 数据源
	 *
	 * @return
	 * @author liangqm
	 * @date 2018-04-24 16:06:18
	 * @params
	 * @since v1.0.1
	 */
	@Bean
	@ConfigurationProperties(prefix = "spring.datasource.batch")
	public DataSource batchDataSource() {
		DruidDataSource batchDataSource = new DruidDataSource();
		return batchDataSource;
	}

	/**
	 * file reader
	 *
	 * @return
	 * @author liangqm
	 * @date 2018-04-24 15:28:18
	 * @params pathToFile 文件路径
	 * @since v1.0.1
	 */
	@Bean
	@StepScope
	public FlatFileItemReader<AccountInfo> reader(@Value("#{jobParameters['input.file.name']}") String pathToFile) {
		FlatFileItemReader<AccountInfo> reader = new FlatFileItemReader<>();
		reader.setResource(new ClassPathResource(pathToFile));
		reader.setLineMapper(new DefaultLineMapper<AccountInfo>() {{
			setLineTokenizer(new DelimitedLineTokenizer() {{
				setNames(new String[]{"user_id", "account_no", "account_name", "deposit_account_no", "amount", "frozen_amount", "account_type", "login_pwd", "draw_pwd", "remark", "create_time", "update_time"});
			}});
			setFieldSetMapper(new BeanWrapperFieldSetMapper<AccountInfo>() {{
				setTargetType(AccountInfo.class);
			}});
		}});
		return reader;
	}


	@Bean
	public ItemProcessor<AccountInfo, AccountInfo> processor() {
		//使用我们自定义的ItemProcessor的实现CsvItemProcessor
		CsvItemProcessor processor = new CsvItemProcessor();
		//为processor指定校验器为CsvBeanValidator()
		processor.setValidator(csvBeanValidator());
		return processor;
	}

	@Bean
	public ItemWriter<AccountInfo> writer(DataSource dataSource) {
		JdbcBatchItemWriter<AccountInfo> writer = new JdbcBatchItemWriter<>();
		//我们使用JDBC批处理的JdbcBatchItemWriter来写数据到数据库
		writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
		String sql = "insert into account_info (user_id, account_no,account_name, deposit_account_no,account_type, login_pwd, draw_pwd,remark, create_time, update_time)" +
				" values ( :userId, :accountNo, :accountName, :depositAccountNo, :accountType, :loginPwd, :drawPwd, :remark, :createTime, :updateTime)";
		//在此设置要执行批处理的SQL语句
		writer.setSql(sql);
		writer.setDataSource(dataSource);
		return writer;
	}

	/**
	 * JobRepository，用来注册Job的容器
	 * jobRepositor的定义需要dataSource和transactionManager，Spring Boot已为我们自动配置了
	 * 这两个类，Spring可通过方法注入已有的Bean
	 *
	 * @param transactionManager
	 * @return
	 * @throws Exception
	 */
	@Bean
	public JobRepository jobRepository(@Qualifier("batchDataSource") DataSource batchDataSource, PlatformTransactionManager transactionManager) throws Exception {
		JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
		jobRepositoryFactoryBean.setDataSource(batchDataSource);
		jobRepositoryFactoryBean.setTransactionManager(transactionManager);
		jobRepositoryFactoryBean.setDatabaseType("mysql");

		JobRepository jobRepository = jobRepositoryFactoryBean.getObject();
		return jobRepository;
	}

	@Bean
	public JobBuilderFactory jobBuilderFactory(JobRepository jobRepository) {
		JobBuilderFactory jobBuilderFactory = new JobBuilderFactory(jobRepository);
		return jobBuilderFactory;
	}


//	@Bean
//	public PlatformTransactionManager transactionManager(@Qualifier("batchDataSource") DataSource dataSource) {
//		return new DataSourceTransactionManager(dataSource);
//	}

	/**
	 * JobLauncher定义，用来启动Job的接口
	 *
	 * @param transactionManager
	 * @return
	 * @throws Exception
	 */
	@Bean
	public SimpleJobLauncher jobLauncher(JobRepository jobRepository, PlatformTransactionManager transactionManager) throws Exception {
		SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
//		jobLauncher.setJobRepository(jobRepository(transactionManager));
		jobLauncher.setJobRepository(jobRepository);
		return jobLauncher;
	}

	/**
	 * Job定义，我们要实际执行的任务，包含一个或多个Step
	 *
	 * @param jobBuilderFactory
	 * @param s1
	 * @return
	 */
	@Bean
	public Job importJob(JobBuilderFactory jobBuilderFactory, Step s1) {

		Job job = jobBuilderFactory.get("importJob")
				.incrementer(new RunIdIncrementer())
				//为Job指定Step
				.flow(s1)
				.end()
				//绑定监听器csvJobListener
				.listener(csvJobListener())
				.build();

		return job;
	}

	/**
	 * step步骤，包含ItemReader，ItemProcessor和ItemWriter
	 *
	 * @param stepBuilderFactory
	 * @param reader
	 * @param writer
	 * @param processor
	 * @return
	 */
	@Bean
	public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<AccountInfo> reader, ItemWriter<AccountInfo> writer, ItemProcessor<AccountInfo, AccountInfo> processor) {
		return stepBuilderFactory
				.get("step1")
				//批处理每次提交100000条数据
				.<AccountInfo, AccountInfo>chunk(100000)
				//给step绑定reader
				.reader(reader)
				//给step绑定processor
				.processor(processor)
				//给step绑定writer
				.writer(writer)
				.build();
	}

	@Bean
	public CsvJobListener csvJobListener() {
		return new CsvJobListener();
	}

	@Bean
	public Validator<AccountInfo> csvBeanValidator() {
		return new CsvBeanValidator<>();
	}

}
